package com.chang.seckill.config;

import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

/**
 * @author: changzhenxi
 * @date: 2024年12月26日 16:19
 * @Version: 1.0
 * @description: 发送端确认  数据可靠消费  setConfirmCallback:消息发布到服务器时的回调确认
 * setReturnsCallback: 服务器的交换机到各个队列时的消息回调确认
 */
@Configuration
public class MyRabbitConfig {


    private RabbitTemplate rabbitTemplate;

    @Primary
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setMessageConverter(messageConverter());
        initRabbitTemplate();
        return rabbitTemplate;
    }

    // 给容器中放一个自定义的消息转换器,使其转换为json格式
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }


    /**
     * 定制Rabbitmq: 保证可靠抵达
     * 1,添加系统配置 publisher-confirm-type: simple   publisher-returns: true
     * 2,设置确认回调
     */
    // @PostConstruct  // 该注解作用就是 在MyRabbitConfig配置类初始化之后，执行该方法  后续备注：这里使用该注解会造成循环依赖
    public void initRabbitTemplate() {

        // 设置发往mq服务器时确认回调   保证消息可靠抵达服务器
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData(重点) 当前消息的唯一关联数据 (消息的唯一id) mq发消息时可以指定对应参数,
             *                            TODO 发消息时可以将该参数保存到数据库中,如果服务端接收到该消息就在数据库中标记一下
             *                            TODO  如果该消息一直没被消费,就可以设置定时任务重新发布 保证数据可靠消费
             * @param ack 消息是否成功收到  只要消息抵达服务器就会是true
             * @param cause 失败原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                // 保证消息可靠需要 消费端手动ack 防止宕机导致消息丢失
                // 将每一个发送的消息都在数据库做好记录，定期将失败的消息再次推送
                // 服务器收到了
                System.out.println("confirm...correlationData[" + correlationData + "]==>ack:[" + ack + "]==>cause:[" + cause + "]");
            }
        });

        // 设置消息打印
        rabbitTemplate.setMandatory(true);

        // 设置消息抵达队列的确认回调  保证消息可靠抵达消息队列
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            /**
             * 只要消息没有投递给指定的队列,触发这个失败回调
             * @param returnedMessage 投递失败的详细信息
             *                         这里报错后就修改数据库当前消息的状态
             */
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                System.out.println("returnedMessage..");
            }
        });

    }
}
